package org.redis.cluster.loadbalance.slot;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.collections.CollectionUtils;
import org.redis.cluster.RedisCluster;
import org.redis.cluster.config.ClientConfig;
import org.redis.cluster.config.ClientInfo;
import org.redis.cluster.config.ClientStatus;
import org.redis.cluster.plugin.Plugin;
import org.redis.cluster.registry.Registry;
import org.redis.cluster.registry.Node;
import org.redis.cluster.util.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SlotManager implements Plugin {
	
	private final static boolean DEFAULT_CACHE_ENABLE=true;

	protected final Logger logger=LoggerFactory.getLogger(getClass());
	
	private final Map<Integer,String> slotCache=new ConcurrentHashMap<Integer,String>();
	
	private RedisCluster redisCluster;
	private Registry registry;
	
	private int slotCount;
	private boolean cacheEnable=DEFAULT_CACHE_ENABLE;
	
	@Override    
	public void init(RedisCluster redisCluster) {
		this.redisCluster=redisCluster;
		this.registry=redisCluster.getRegistry();
		this.slotCount=redisCluster.getConfigManager().getSystemConfig().get(SlotConstants.SLOT_COUNT_SYSTEM_CONFIG).getValue(Integer.class, SlotConstants.DEFAULT_SLOT_COUNT);
		Assert.isTrue(slotCount>0,"%s必须大于0",SlotConstants.SLOT_COUNT_SYSTEM_CONFIG);
		
		initSlot();
		initSlotListener();
	}
	
	protected void initSlot() {
		String slotInitedNode=SlotHelper.getSlotPath(SlotConstants.INITED_NODE);
		registry.register(SlotHelper.getSlotPath(""));
		registry.register(SlotHelper.getSlotPath(SlotConstants.INDEX_PATH));
		registry.register(SlotHelper.getSlotPath(SlotConstants.LEADER_PATH));
		registry.register(SlotHelper.getSlotPath(SlotConstants.MAPPING_PATH));
		
		if(registry.isExist(slotInitedNode)) {
			return;
		}
		
		registry.register(slotInitedNode, null);
		
		if(logger.isInfoEnabled()) {
			logger.info("正在初始化{}个槽数据",slotCount);
		}
		
		try {
			for(int i=0;i<slotCount;i++) {
				registry.register(SlotHelper.getSlotIndexPath(i));
			}
			
			if(logger.isInfoEnabled()) {
				logger.info("初始化槽数据完成");
			}
		}
		catch(Exception e) {
			if(logger.isErrorEnabled()) {
				logger.error("初始化槽数据失败 - ",e);
			}
			
			registry.unregister(slotInitedNode);
		}
	}
	
	protected void initSlotListener() {
		registry.listen(SlotHelper.getSlotPath(SlotConstants.INDEX_PATH), (String path)->{
			slotCache.clear();
		});
	}
	
	@Override
	public void destroy() {
	}
	
	/**
	 * 注册
	 * @param clientConfig
	 * @return
	 */
	public boolean registerSlotClient(SlotClientConfig slotClientConfig) {
		ClientConfig clientConfig=SlotClientConfigAdapter.adaptClientConfig(slotClientConfig);
		slotClientConfig.setId(clientConfig.getId());
		
		ClientInfo clientInfo=new ClientInfo();
		clientInfo.setStatus(ClientStatus.INIT);
		clientConfig.setInfo(clientInfo);
		
		registry.registerClient(clientConfig);
		return registry.register(SlotHelper.getSlotReplicatePath(clientConfig.getId()),new Node(slotClientConfig));
	}
	
	/**
	 * 取消注册
	 * @param sourceSlotClientConfig
	 * @param targetSlotClientConfig
	 * @return
	 */
	public boolean unregisterSlotClient(SlotClientConfig sourceSlotClientConfig,SlotClientConfig targetSlotClientConfig) {
		SlotMapping sourceSlotMapping=getSlotMapping(sourceSlotClientConfig.getId());
		if(sourceSlotMapping!=null) {
			if(CollectionUtils.isEmpty(sourceSlotMapping.getSlots())) {
				return false;
			}
			
			targetSlotClientConfig.setSlots(sourceSlotMapping.getSlots());
		}
		
		return registerSlotClient(targetSlotClientConfig);
	}
	
	/**
	 * @param clientId
	 * @return
	 */
	public SlotMapping getSlotMapping(String clientId) {
		Node node=registry.getNode(SlotHelper.getSlotMappingPath(clientId));
		if(node==null) {
			return null;
		}
		
		return node.getData();
	}
	
	/**
	 * @param slotMapping
	 * @return
	 */
	protected boolean setSlotMapping(SlotMapping slotMapping) {
		String path=SlotHelper.getSlotMappingPath(slotMapping.getClientId());
		Node node=new Node(slotMapping);
		if(registry.register(path, node)) {
			return true;
		}
		
		return registry.setNode(path, node);
	}
	
	/**
	 * @param clientId
	 * @return
	 */
	public boolean removeSlotMapping(String clientId) {
		return registry.unregister(SlotHelper.getSlotMappingPath(clientId));
	}

	/**
	 * 
	 * @return
	 */
	public Collection<SlotClientConfig> getSlotClientConfigs() {
		Collection<Node> nodes=registry.getNodes(SlotHelper.getSlotPath(SlotConstants.REPLICATE_PATH));
		if(CollectionUtils.isEmpty(nodes)) {
			return Collections.emptyList();
		}

		Collection<SlotClientConfig> result=new ArrayList<SlotClientConfig>(nodes.size());
		for(Node node:nodes) {
			Node configNode=registry.getNode(SlotHelper.getSlotReplicatePath(node.getPath()));
			if(configNode!=null) {
				result.add(configNode.getData());
			}
		}
		
		return result;
	}
	
	/**
	 * 
	 * @return
	 */
	public SlotClientConfig getReplicatingSlotClientConfig() {
		Collection<Node> nodes=registry.getNodes(SlotHelper.getSlotPath(SlotConstants.REPLICATE_PATH));
		if(CollectionUtils.isEmpty(nodes)) {
			return null;
		}
		
		for(Node node:nodes) {
			Node configNode=registry.getNode(SlotHelper.getSlotReplicatePath(node.getPath()));
			if(configNode!=null) {
				return configNode.getData();
			}
		}

		return null;
	}

	/**
	 * 取消注册
	 * @param config
	 * @return
	 */
	public boolean unregisterSlotClifngConfig(SlotClientConfig config) {
		return registry.unregister(SlotHelper.getSlotReplicatePath(config.getId()));
	}
	 
	/**
	 * 返回
	 * @param index
	 * @return
	 */
	public String getSlotClientId(int index) {
		if(slotCache.containsKey(index)) {
			return slotCache.get(index);
		}
		
		Node node=registry.getNode(SlotHelper.getSlotIndexPath(index));
		String clientId="";
		if(node!=null && node.getData()!=null) {
			clientId=node.getData();
		}
		
		if(isCacheEnable()) {
			slotCache.put(index, clientId);
		}
		
		return clientId;
	}
	
	/**
	 * 设置
	 * @param index
	 * @param clientId
	 * @return
	 */
	public boolean setSlotClientId(int index,String clientId) {
		String path=SlotHelper.getSlotIndexPath(index);
		boolean result=registry.setNode(path, new Node(clientId));
		
		if(result) {
			slotCache.remove(index);
		}
		return result;
	}
	
	/**
	 * 返回SlotCount
	 * @return
	 */
	public int getSlotCount() {
		if(slotCount>0) {
			return slotCount;
		}
		
		Collection<Node> nodes=registry.getNodes(SlotHelper.getSlotPath(SlotConstants.INDEX_PATH));
		if(CollectionUtils.isEmpty(nodes)) {
			slotCount=0;
			return slotCount;
		}
		
		slotCount=nodes.size();
		return slotCount;
	}
	
	public void setRedisCluster(RedisCluster redisCluster) {
		this.redisCluster = redisCluster;
	}
	public RedisCluster getRedisCluster() {
		return redisCluster;
	}
	public void setRegistry(Registry registry) {
		this.registry = registry;
	}
	public Registry getRegistry() {
		return registry;
	}
	public void setCacheEnable(boolean cacheEnable) {
		this.cacheEnable = cacheEnable;
	}
	public boolean isCacheEnable() {
		return cacheEnable;
	}
}